package com.tony.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @version V1.0  </br>
 * @ClassName KafkaProducer </br>
 * @Description TODO  </br>
 * @Author asus</br>
 * @DATE 2019/3/7  19:57</br>
 */
public class Producer {
    private  final KafkaProducer<Integer, String> producer;
    private final String topic;
    private final boolean isAysnc;

    public Producer(String topic, boolean isAysnc) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.17.10.185:9092,172.17.10.186:9092,172.17.10.187:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerTest");
        properties.put(ProducerConfig.ACKS_CONFIG, "-1");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<Integer, String>(properties);
        this.topic = topic;
        this.isAysnc = isAysnc;
    }

    public  void start() {
        new Thread(
                () ->{
                    for (int i=0;i < Integer.MAX_VALUE; i++) {
                        String message = "message_" + i;
                        System.out.println("begin send message:" + message);
                        if (isAysnc) { //异步发送
                            producer.send(new ProducerRecord<Integer, String>(topic, message), (recordMetadata, e) ->{
                                if (recordMetadata != null) {
                                    System.out.println("async--offset:" + recordMetadata.offset() +
                                            "->partition:" + recordMetadata.partition());
                                }
                            });
                        }else {// 同步发送
                            RecordMetadata recordMetadata = null;
                            try {
                                recordMetadata = producer.send(new ProducerRecord<Integer, String>(topic , message)).get();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            } catch (ExecutionException e) {
                                e.printStackTrace();
                            }
                            System.out.println("sync-offset:" + recordMetadata.offset() +
                                    "->partition:" + recordMetadata.partition());
                        }
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
        ).start();
    }
}
